package cn.com.open.itembank.appender;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author 奶茶
 * Created by on 2018/5/24.
 */
public abstract class DaemonAppender<E> implements Runnable {
    private ExecutorService threadPool = Executors.newCachedThreadPool();

    private static final Logger log = LoggerFactory.getLogger(DaemonAppender.class);

    private AtomicBoolean start = new AtomicBoolean(false);
    private final BlockingQueue<E> queue;

    DaemonAppender(int maxQueueSize) {
        this.queue = new LinkedBlockingQueue<>(maxQueueSize);
    }

    protected void execute() {
        if (threadPool.isShutdown()) {
            threadPool = Executors.newCachedThreadPool();
        }
        threadPool.execute(this);
    }

    void log(E eventObject) {
        if (!queue.offer(eventObject)) {
            log.warn(String.format("Message queue is full. Ignored the message: %s%s", System.lineSeparator(), eventObject.toString()));
        } else if (start.compareAndSet(false, true)) {
            execute();
        }
    }

    public void run() {
        try {
            append(queue.take());
        } catch (Exception e) {
            close();
        }
    }

    protected abstract void append(E rawData);

    protected void close() {
        synchronized (threadPool) {
            if (!threadPool.isShutdown()) {
                shutdownAndAwaitTermination(threadPool);
            }
        }
    }

    private static void shutdownAndAwaitTermination(ExecutorService pool) {
        pool.shutdown();
        try {
            if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
                pool.shutdownNow();
                if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
                    log.info("Pool did not terminate");
                }
            }
        } catch (InterruptedException ie) {
            pool.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}
